package com.carol.bigdata

import java.sql.Timestamp
import java.util.concurrent.{Callable, Executors, FutureTask}

import com.carol.bigdata.fuction.MyAccumulator
import com.carol.bigdata.kudu.KuduUtil
import com.carol.bigdata.utils.TimeUtil
import com.carol.bigdata.kudu.KuduUtil
import com.carol.bigdata.utils.TimeUtil
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}



object Task {
    
    // 建表的固定字段Schema定义
    val userInfoSchema: StructType = StructType(
        StructField("uid", StringType, false) ::
            StructField("uid_type", StringType, false) ::
            StructField("game_id", IntegerType, false) ::
            StructField("day", StringType, false) ::
            StructField("agent", StringType, true) ::
            StructField("ip", StringType, true) ::
            StructField("timestamp", TimestampType, true) ::
            StructField("time", TimestampType, true) ::
            StructField("timezone", StringType, true) ::
            StructField("year", StringType, true) ::
            StructField("month", StringType, true) ::
            StructField("week", StringType, true) ::
            StructField("hour", StringType, true) ::
            StructField("minute", StringType, true) :: Nil
        )
    // 固定表主键
    val keySeq = Seq("uid", "uid_type", "game_id")
    // 固定分区键
    val partitionList = List("uid_type", "game_id")
    
    /*
        Map(channel_id -> 1, device_no -> device_2, event -> sdk_download, device -> oppo, game_id -> 1, subchannel_id -> 1, platform_id -> 1, timestamp -> 1601506800, media_id -> 1)
        Map(channel_id -> 1, device_no -> device_5, event -> sdk_download, device -> vivo, game_id -> 1, subchannel_id -> 1, platform_id -> 1, timestamp -> 1601593200, media_id -> 1)
        Map(channel_id -> 1, device_no -> device_13, event -> sdk_download, device -> huawei, game_id -> 1, subchannel_id -> 1, platform_id -> 1, timestamp -> 1602111600, media_id -> 1)
    */
    // 单线程写入多表
    def run(kuduContext: KuduContext,
            spark: SparkSession,
            data: RDD[(String,Map[String, (String,String)])],
            myAccumulator: MyAccumulator): Unit = {
        val gameSet = myAccumulator.value
        gameSet.foreach(game_id => {
            println(s"==写入game_id为$game_id==")
            // 过滤出当前game_id的数据
            val valueRDD = data.filter(x=> x._2.getOrElse("game_id", ("-1",""))._1.toInt == game_id)
            val tableName = "user_profile_" + game_id
            // 如果表不存在,则会创建定义好的固定字段的表
            KuduUtil.createTable(kuduContext, tableName, userInfoSchema, keySeq, partitionList, 2)
            // 动态获取表结构
            val (schema, intStampFields, doubleFields, floatFields, boolFields, timeStampFields, columnList) = KuduUtil.getSchema(kuduContext, tableName)
            // 获取kudu历史表数据 RDD[(uid_uidType,map)]
            val historyRDD: RDD[(String, Map[String, String])] = KuduUtil.readKudu(spark, tableName,kuduContext.kuduMaster).rdd.map(row => {
                val uid = row.getAs[String]("uid")
                val uid_type = row.getAs[String]("uid_type")
                val map = row.getValuesMap[String](columnList)
                (uid+"-"+uid_type,map)
            })
            
            // 合并当前批次的最新数据
            val mergeRDD: RDD[(String, Map[String, String])] = valueRDD.reduceByKey((map1, map2) => {
                (map1 /: map2) ((map, kv) => {
                    // 获取最新时刻的信息
                    val (value, time) = if (map.getOrElse(kv._1, ("", "0000-00-00 00:00:00.000"))._2 > kv._2._2) map.getOrElse(kv._1, ("", "0000-00-00 00:00:00.000")) else kv._2
                    map + (kv._1 -> (value, time))
                })
            }).map(x => (x._1, x._2.mapValues(_._1)))
            
            // 关联用户历史数据，进行数据合并
            val combineRDD: RDD[Map[String, String]] = mergeRDD.leftOuterJoin(historyRDD).map(x => {
                // 历史数据
                val oldMap: Map[String, String] = x._2._2.getOrElse(Map[String, String]())
                // 最新数据
                val newMap = x._2._1
                if (oldMap.nonEmpty) {
                    // 合并map,将用户历史消息添加到newMap中
                    combineMap(newMap, oldMap)
                } else {
                    newMap
                }
            })
            
            // 按表字段排序，生成DataFrame
            val resultRDD: RDD[Row] = getResultRDD(combineRDD, intStampFields, doubleFields, floatFields, boolFields, timeStampFields, columnList)
            val resultDF: DataFrame = spark.createDataFrame(resultRDD, schema)
            
            // 写入kudu对应的game表
            KuduUtil.upsertTable2(tableName, resultDF, kuduContext.kuduMaster)
            println(s"==game_id为$game_id 写入完成==")
        })
    }
    
    // 多线程写入,每个线程处理一个游戏表
    def multiThreadRun(kuduContext: KuduContext,
                       spark: SparkSession,
                       data: RDD[(String,Map[String, (String,String)])],
                       myAccumulator: MyAccumulator,
                       threadPool: Int = 3): Unit = {
        val gameSet = myAccumulator.value
        println("游戏集合：" + gameSet)
        val jobExecutor = Executors.newFixedThreadPool(threadPool)
        var tableList = List[(String, DataFrame)]()
        gameSet.foreach(game_id => {
            // 过滤出当前game_id的数据
            val valueRDD = data.filter(x=> x._2.getOrElse("game_id", ("-1",""))._1.toInt == game_id)
            val tableName = "user_profile_" + game_id
            // 如果表不存在,则会创建定义好的固定字段的表
            KuduUtil.createTable(kuduContext, tableName, userInfoSchema, keySeq, partitionList, 2)
            // 动态获取表结构
            val (schema, intStampFields, doubleFields, floatFields, boolFields, timeStampFields, columnList) = KuduUtil.getSchema(kuduContext, tableName)
            // 获取kudu历史表数据 RDD[(uid_uidType,map)]
            val historyRDD: RDD[(String, Map[String, String])] = KuduUtil.readKudu(spark, tableName, kuduContext.kuduMaster).rdd.map(row => {
                val uid = row.getAs[String]("uid")
                val uid_type = row.getAs[String]("uid_type")
                val map = row.getValuesMap[String](columnList)
                (uid+"-"+uid_type,map)
            })
    
            // 合并当前批次的最新数据
            val mergeRDD: RDD[(String, Map[String, String])] = valueRDD.reduceByKey((map1, map2) => {
                (map1 /: map2) ((map, kv) => {
                    // 获取最新时刻的信息
                    val (value, time) = if (map.getOrElse(kv._1, ("", "0000-00-00 00:00:00.000"))._2 > kv._2._2) map.getOrElse(kv._1, ("", "0000-00-00 00:00:00.000")) else kv._2
                    map + (kv._1 -> (value, time))
                })
            }).map(x => (x._1, x._2.mapValues(_._1)))
    
            // 关联用户历史数据，进行数据合并
            val combineRDD: RDD[Map[String, String]] = mergeRDD.leftOuterJoin(historyRDD).map(x => {
                // 历史数据
                val oldMap: Map[String, String] = x._2._2.getOrElse(Map[String, String]())
                // 最新数据
                val newMap = x._2._1
                if (oldMap.nonEmpty) {
                    // 合并map,将用户历史消息添加到newMap中
                    combineMap(newMap, oldMap)
                } else {
                    newMap
                }
            })
            
            // 按表字段排序，生成DataFrame
            val resultRDD: RDD[Row] = getResultRDD(combineRDD, intStampFields, doubleFields, floatFields, boolFields, timeStampFields, columnList)
            val resultDF: DataFrame = spark.createDataFrame(resultRDD, schema)
            tableList +:= (tableName, resultDF)
        })
        println("table集合：" + tableList.size)
        // 7个game、threadPool=3   1、2、3
        for (epoch <- 1 to math.ceil(gameSet.size.toDouble / threadPool.toDouble).toInt) {
            var futureList = List[FutureTask[Int]]()
            for (t <- 1 to threadPool) {
                val i = (epoch - 1) * threadPool + t - 1 // i=0、1、2、3、4、5、6、7、8
                if (i < tableList.size) { // size=7
                    val (tableName, resultDF) = tableList(i)
                    val future: FutureTask[Int] = new FutureTask[Int](new Callable[Int] {
                        override def call(): Int = {
                            // 写入kudu对应的game表
                            KuduUtil.upsertTable2(tableName, resultDF, kuduContext.kuduMaster)
                            println(s"==$tableName 写入完成==")
                            1
                        }
                    })
                    futureList :+= future
                }
            }
            try {
                // 每threadPool个任务提交一次
                for (future <- futureList) {
                    jobExecutor.execute(future)
                }
            } catch {
                case exception: Exception => {
                    //关闭线程
                    exception.printStackTrace()
                    jobExecutor.shutdown()
                }
            }
        }
        jobExecutor.shutdown()
        try {
            // awaitTermination返回false即超时会继续循环，返回true即线程池中的线程执行完成主线程跳出循环往下执行，每隔10秒循环一次
            while (!jobExecutor.awaitTermination(10, java.util.concurrent.TimeUnit.SECONDS)) {
                println("还有后台线程在执行，阻塞主线程！！！")
            }
        }
        catch {
            case e: InterruptedException => e.printStackTrace();
        }
    }
    
    def getResultRDD(valueRDD: RDD[Map[String, String]],
                     intFields: List[String],
                     doubleFields: List[String],
                     floatFields: List[String],
                     boolFields: List[String],
                     timeStampFields: List[String],
                     columnList: List[String]): RDD[Row] = {
        valueRDD.map(x => {
            var valueList = List[Any]()
            // 获取时间字段
            val timestampValue = x.getOrElse("time", "0")
            val timezone = x.getOrElse("timezone", "Asia/Shanghai")
            val timeValue = TimeUtil.timestamp2str(timestampValue, timezone = timezone)
            for (c <- columnList) {
                if (intFields.contains(c)) {
                    valueList :+= x.getOrElse(c, "0").toInt
                } else if (timeStampFields.contains(c)) {
                    if (c.equals("time")) valueList :+= Timestamp.valueOf(timeValue)
                    else valueList :+= Timestamp.valueOf(TimeUtil.timestamp2str(x.getOrElse(c, "0"), timezone = timezone))
                } else if (doubleFields.contains(c)) {
                    valueList :+= x.getOrElse(c, "0").toDouble
                } else if (floatFields.contains(c)) {
                    valueList :+= x.getOrElse(c, "0").toFloat
                } else if (boolFields.contains(c)) {
                    valueList :+= x.getOrElse(c, "false").toBoolean
                } else {
                    var value = x.getOrElse(c, "")
                    if (c.equals("year")) {
                        value = timeValue.take(4)
                    } else if (c.equals("month")) {
                        value = timeValue.take(7)
                    } else if (c.contains("day")) {
                        value = timeValue.take(10)
                    } else if (c.equals("hour")) {
                        value = timeValue.take(13)
                    } else if (c.equals("minute")) {
                        value = timeValue.take(16)
                    } else if (c.equals("week")) {
                        value = timeValue.take(10) + ":" + timeValue.substring(11, 13) + getWeekOfMonth(timeValue)
                    }
                    valueList :+= value
                }
            }
            Row.fromSeq(valueList)
        })
    }
    
    def getWeekOfMonth(day: String): String = {
        val num = (day.substring(8, 10).toInt - 1) / 7 + 1
        val weekOfMonth = num match {
            case 1 => "(第一周)"
            case 2 => "(第二周)"
            case 3 => "(第三周)"
            case 4 => "(第四周)"
            case 5 => "(第五周)"
        }
        weekOfMonth
    }
    
    // 将oldMap中有而newMap中没有的kv添加到newMap中
    def combineMap(newMap: Map[String, String],
                   oldMap: Map[String, String]): Map[String, String] ={
        val oldSet = oldMap.keySet
        val newSet = newMap.keySet
        var map = newMap
        val keySet = oldSet -- newSet
        keySet.foreach(key => {
            map += (key -> oldMap.getOrElse(key,"null"))
        })
        map
    }
    
    def main(args: Array[String]): Unit = {
        //val timeValue = "2021-12-31 03:52:11"
        //val week = timeValue.take(10) + ":" + timeValue.substring(11, 13) + getWeekOfMonth(timeValue)
        //println(week)
        //println(getWeekOfMonth("2021-12-07"))
        //println(getWeekOfMonth("2021-12-31"))
    
        val map = Map[String, String]("test1" -> "111")
        val map2 = Map[String, String]("test2" -> "222")
    
        val stringToString = combineMap(map, map2)
        println(stringToString)
        
    }
    
}
